package com.gin.monitor.agg.connector.kafka.util;

import com.gin.monitor.agg.vo.TrafficInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.nio.charset.StandardCharsets;

/**
 * @author gin
 * @date 2021/4/23
 */
public class TrafficInfoDeserializationSchema implements KafkaDeserializationSchema<TrafficInfo> {

    private static final long serialVersionUID = -7414144415839064441L;

    @Override
    public boolean isEndOfStream(TrafficInfo nextElement) {
        //判断结束
        return false;
    }

    @Override
    public TrafficInfo deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) {
        //要进行序列化的字节流
        //String key = new String(consumerRecord.key(), StandardCharsets.UTF_8)
        if (null == consumerRecord || null == consumerRecord.value()) {
            return null;
        } else {
            String trafficInfoStr = new String(consumerRecord.value(), StandardCharsets.UTF_8);
            String[] split = trafficInfoStr.split(",");
            return new TrafficInfo(Long.parseLong(split[0]), split[1], split[2],
                    split[3], Double.parseDouble(split[4]), split[5], split[6]);
        }
    }

    @Override
    public TypeInformation<TrafficInfo> getProducedType() {
        //指定一下返回的数据类型  Flink提供的类型
        return TypeInformation.of(new TypeHint<TrafficInfo>() {
        });
    }

}